package com.yl.flink.processor;

import cn.hutool.core.date.DateUtil;
import com.yl.constant.Const;
import com.yl.entity.MultiDataEntity;
import com.yl.util.DtUtil;
import com.yl.util.SUtil;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author wlf
 * @since 2022/8/18
 */
public class OnlineStatusMapFunc extends KeyedProcessFunction<String, MultiDataEntity, Tuple3<String, Integer, Integer>> {

    private final String projectName;
    // 每个测点id对应的最近一次更新redis状态的时间戳
    private transient MapState<String, Long> latestTsOfUpdate;

    public OnlineStatusMapFunc(ParameterTool params) {
        this.projectName = params.get(Const.APP_NAME);
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        latestTsOfUpdate = getRuntimeContext().getMapState(new MapStateDescriptor<>(Const.S_STATUS_ONLINE, BasicTypeInfo.STRING_TYPE_INFO, TypeInformation.of(new TypeHint<>() {
        })));
    }

    /**
     * 测点状态(-1离线，0在线，1紧急，2严重，3一般)
     * 这里把在线的状态缓存到redis，即key为测点id，value为0
     * 注意：高频数据最快一秒更新一次缓存
     */
    @Override
    public void processElement(MultiDataEntity multiDataEntity, KeyedProcessFunction<String, MultiDataEntity, Tuple3<String, Integer, Integer>>.Context ctx, Collector<Tuple3<String, Integer, Integer>> out) throws Exception {
        String redisKey = SUtil.fmtRedisKey(projectName, Const.REDIS_KEY_STATUS_ONLINE, multiDataEntity.getTarget().getId());
        int interval = multiDataEntity.getInterval();
        if (latestTsOfUpdate.contains(redisKey)) {
            // 如果缓存中有该key,说明更新过状态，那就需要判断更新间隔是否满足再次更新的条件
            if (DtUtil.diffTime(latestTsOfUpdate.get(redisKey), DateUtil.date().getTime()) > interval) {
                outputAndUpdate(redisKey, interval, out);
            }
        } else {
            // 如果缓存中还没有该key，说明还没更新过状态，那就更新一次，同时把更新的时间戳存起来
            outputAndUpdate(redisKey, interval, out);
        }
    }

    /**
     * 输出到下游更新redis中的测点状态，同时把这次更新的时间缓存起来
     */
    private void outputAndUpdate(String redisKey, int interval, Collector<Tuple3<String, Integer, Integer>> out) throws Exception {
        latestTsOfUpdate.put(redisKey, DateUtil.date().getTime());
        out.collect(Tuple3.of(redisKey, 0, interval));
    }

}
